package com.cloud.mq.consumer;

import com.cloud.common.core.constant.SecurityConstants;
import com.cloud.common.core.utils.JsonUtil;
import com.cloud.system.api.RemoteLogService;
import com.cloud.system.api.domain.SysOperLog;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

/**
 * Kafaka消费端
 */
@Service
public class KafkaConsumer {

    @Autowired
    private RemoteLogService remoteLogService;

//    @KafkaListener(topics = "log_queue", groupId = "zq-group")
    @KafkaListener(topics = "log_queue", groupId = "${spring.application.name}-group")
    public void saveLog(String msg) throws Exception {
        SysOperLog sysOperLog = JsonUtil.jsonToBean(msg, SysOperLog.class);
        remoteLogService.saveLog(sysOperLog, SecurityConstants.INNER);
//        System.out.println("收到消息："+msg);
    }

    private final String topicName = "hub-topic-city-delta";
//    监听消息队列，实现自动消费消息队列
    @KafkaListener(
            topics = {topicName},
            groupId = "hub-topic-city-delta-group-aaa")
    public void consumeMsg(String msg) {
        try {
            System.out.println("收到消息aaa："+msg);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    @KafkaListener(
            topics = {topicName},
            groupId = "hub-topic-city-delta-group-ddd")
    public void consumeMsgdd(String msg) {
        try {
            System.out.println("收到消息ddd："+msg);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
